package com.shujia.spark.streaming


import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, SparkSession}
import org.apache.spark.streaming.{Durations, StreamingContext}
import org.apache.spark.streaming.dstream.DStream

object Demo04Transform {
  def main(args: Array[String]): Unit = {
    val spark: SparkSession = SparkSession
      .builder()
      .appName(this.getClass.getSimpleName.replace("$", ""))
      .master("local[2]") // 至少需要两个线程，因为接收数据需要一致占用一个线程
      .config("spark.default.parallelism", "2")
      .getOrCreate()

    // 构建StreamingContext：需要传入SparkContext以及Duration表示批次大小
    val ssc: StreamingContext = new StreamingContext(spark.sparkContext, Durations.seconds(5))

    // 通过Socket连接 从nc服务中加载数据
    // Socket服务端：nc -lk 8888
    // 会返回DStream类型，SparkStreaming的一种编程模型
    val lineDS: DStream[String] = ssc.socketTextStream("master", 8888)

    import spark.implicits._
    import org.apache.spark.sql.functions._

    lineDS
      // 将基于DStream的流处理转换成RDD处理
      // 将指定的5s的数据封装成一个批次转换成RDD进行处理
      .transform((rdd: RDD[String]) => {
        // 对RDD进行转换，最终需要返回一个新的RDD
        val kvRDD: RDD[(String, Int)] = rdd
          .flatMap(_.split(","))
          .map(w => (w, 1))
          .reduceByKey(_ + _)

        // RDD是可以转换成DF
        val df: DataFrame = kvRDD.toDF("word", "cnt")

        // 通过SQL完成需求


        df.rdd
      })
    //      .print()

    lineDS
      .transform(rdd => {
        val lineDF: DataFrame = rdd.map(s => Tuple1(s)).toDF("line")

        lineDF.createOrReplaceTempView("words")

        val wordCnfDF: DataFrame = spark.sql(
          """
            |select  t1.word
            |       ,count(*) as cnt
            |from (
            |select explode(split(line,",")) as word
            |from words
            |) t1 group by t1.word
            |""".stripMargin)

        val wordCntDSLDF: DataFrame = lineDF
          .select(explode(split($"line", ",")) as "word")
          .groupBy($"word")
          .agg(count("*") as "cnt")

        wordCntDSLDF.rdd
      })
      .print()

    // 启动Streaming任务
    ssc.start()
    ssc.awaitTermination() // 等待程序中断

  }

}
